package com.edata.bigdata.flink;

import com.edata.bigdata.basic.Commons;
import com.edata.bigdata.flink.annotations.Key;
import org.apache.flink.api.java.functions.KeySelector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.lang.reflect.Field;

public class EventKeyExtractor<T> implements KeySelector<T, String> {
    public Logger logger = LoggerFactory.getLogger(this.getClass());

    private final Class<T> clazz;

    @Override
    public String getKey(T evt) {
        try {
            Field field = Commons.findFieldFromBeanByAnnotation(clazz, Key.class);
            field.setAccessible(true);
            Object key = field.get(evt);
            return key != null ? key.toString() : "null-key";
        } catch (Exception e) {
            String key = "unknown-key";
            logger.warn("无法找到 {} 的 key 字段，设置为：{}", clazz, key);
            return key;
        }
    }

    public EventKeyExtractor(Class<T> clazz) {
        this.clazz = clazz;
    }
}
